//========================================================================
/**
 * Author   : cuisw <shaovie@gmail.com>
 * Date     : 2008-10-17 23:42
 */
//========================================================================

#ifndef _EPOLLPROACTOR_H_
#define _EPOLLPROACTOR_H_
#include "Pre.h"

#include "Pipe.h"
#include "Token.h"
#include "TaskWithMQ.h"
#include "ProactorImpl.h"
#include "RecursiveThreadMutex.h"
#include "ConditionThreadMutex.h"

class AsynchHandler;
class EpollAsynchOpt;
class EpollAsynchResult;

/**
 * @class EpollProactorNotify 
 *
 * @brief 
 */
class EpollProactorNotify : public AsynchHandler
{
public:
    //
    int open (void);

    //
    int close (void);

    //
    int notify (void);

    //
    NDK_HANDLE notify_handle (void) const;

    // Handle read stream
    void handle_read_stream (void);
protected:
    Pipe notification_pipe_;
};
/**
 * @class EpollProactorToken
 *
 * @brief 
 */
class EpollProactor;
class EpollProactorToken : public Token
{
public:
    EpollProactorToken (EpollProactor *p,
	    int s_queue = Token::FIFO);

    virtual ~EpollProactorToken();

    // Get the proactor implementation
    EpollProactor *proactor (void);

    // Set the proactor implementation
    void proactor (EpollProactor *);   

    // 
    virtual void sleep_hook (void);
private:
    // Proactor implement
    EpollProactor *proactor_;
};
// --------------------------------------------------------------------------
/**
 * @class EpollProactorWriteTask
 *
 * @brief This class is defined for emulating Asynchronous write operation.
 */
class EpollProactorWriteTask : public TaskWithMQ
{
public:
    enum
    {
        AIO_DEL_WRITE_JOB   = 0x911,

        AIO_ADD_WRITE_JOB   = 0x912
    };
    class JobInfo
    {
    public:
        JobInfo ()
        : handle (NDK_INVALID_HANDLE)
        , operation (0)
        , result (0)
        {}

        NDK_HANDLE handle;

        //
        EpollAsynchOpt *operation;

        EpollAsynchResult *result;
    };
    //
    EpollProactorWriteTask ();

    //
    ~EpollProactorWriteTask ();

    //
    virtual int svc (void);

    // Get a free message block from message block pool.
    MessageBlock *get_message_block (void);
protected:
    MessageBlock *get_message_block_i (void);

    // Unget a message block to message block pool.
    void unget_message_block (MessageBlock *);

    void unget_message_block_i (MessageBlock *);

    int execute_write_operation (void);

    void check_message_block (MessageBlock *mb);

    //
    MessageBlock *free_list_head_;

    MessageBlock *free_list_tail_;

    ThreadMutex free_list_mutex_;

    //
    MessageBlock *job_list_head_;

    MessageBlock *job_list_tail_;

    //
    size_t free_count_;

    size_t size_;

    volatile int task_free_;
};
// --------------------------------------------------------------------------
/**
 * @class EpollProactor
 *
 * @brief Asynchronous I/O for epoll actually is non-blocking I/O. We 
 * simulate AIO use epoll bases Edge Triggered model. For writing operation
 * we spawn a single Task to simulate asynchronous writing, and You can setup 
 * the number of thread serve for this `Task'. But for the epoll event 
 * distribution mechanism  we use single thread. 
 */
class EpollProactor : public ProactorImpl
{
public:
    //
    EpollProactor ();

    ~EpollProactor ();

    /**
     * Initialize EpollProactor with size.
     * Note On Unix platforms, the <max_handles> parameter should be 
     * as large as the maximum number of file descriptors allowed 
     * for a given process.  This is necessary since a file descriptor
     * is used to directly index the array of event handlers
     * maintained by the Proactor's handler repository.  Direct
     * indexing is used for efficiency reasons.  If the size
     * parameter is less than the process maximum, the process
     * maximum will be decreased in order to prevent potential
     * access violations.
     * max_size_ whill be set to ::max_handles () if <max_handles> 
     * equel to zero
     * <num_of_threads> must greater then two, We fix the number of 
     * thread that epoll uses, other threads is used to write task. 
     */
    int open (int num_of_threads = 2, int max_handles = 0);

    int close (void);

    void owner (thread_t thr_id);

    //
    int notify (void);

    /**
     * Dispatch a single set of events. If <time_out> elapses
     * before any events occur, return 0. Return 1 if a completion
     * dispatched. Return -1 on errors.
     */
    virtual int handle_events (const TimeValue *timeout = 0);

    // = Register an handle to proactor.
    virtual int register_handle (NDK_HANDLE handle,
	    const void *completion_key,
	    int event_mask);

    int remove_handle (NDK_HANDLE handle, 
	    const void *completion_key,
	    int event_mask);

    //
    virtual int run_event_loop (void);

    //
    virtual void end_event_loop (void);

    //
    virtual int event_loop_done (void);

    // = Methods used to create Asynch IO factory and result objects. We
    // create the right objects here in these methods.
    virtual AsynchReadStreamImpl *create_asynch_read_stream (void);
    virtual AsynchReadStreamResultImpl *create_asynch_read_stream_result (
	    const AsynchHandler *handler,
	    NDK_HANDLE handle, 
	    MessageBlock *mb,
	    size_t bytes_to_read);

    //
    virtual AsynchWriteStreamImpl *create_asynch_write_stream (void);
    virtual AsynchWriteStreamResultImpl *create_asynch_write_stream_result (
	    const AsynchHandler *handler,
	    NDK_HANDLE handle,
	    MessageBlock *mb,
	    size_t bytes_to_write);

    //
    virtual AsynchAcceptImpl *create_asynch_accept (void);
    virtual AsynchAcceptResultImpl *create_asynch_accept_result (
	    const AsynchHandler *handler,
	    NDK_HANDLE listen_handle,
	    MessageBlock *mb,
	    size_t num_to_accept);

    // no use
    virtual int cancel_aio (NDK_HANDLE handle, int evet_mask);

    virtual size_t number_of_threads (void) const;

    virtual void number_of_threads (size_t threads);

    // = Special methods.
    int start_read (EpollAsynchOpt *opt_impl);

    int start_write (EpollAsynchOpt *opt, EpollAsynchResult *result);
protected:
    /**
     * @class EpollAioFdInfo
     *
     * @brief .
     */
    class EpollAioFdInfo
    {
    public:
	enum
	{
	    FD_CLOSED	     = 0x110,

	    FD_REGISTERED    = 0x111,

	    FD_FINISHED	     = 0x112,

	    FD_NOT_FINISHED  = 0x113
	};
	// Construtor.
	EpollAioFdInfo ();

	void reset (void);

	// = Members.
	
	NDK_HANDLE  handle;

	int event_mask;

	int active_mask;

	volatile int status;

	EpollAsynchOpt  *operation;

	EpollAioFdInfo *prev;

	EpollAioFdInfo *next;
    };
    //
    int handle_event_i (const TimeValue *timeout);
    
    //
    int register_handle_i (NDK_HANDLE handle,
	    const void *completion_key,
	    int event_mask);

    int remove_handle_i (NDK_HANDLE handle, 
	    const void *completion_key,
	    int event_mask);

    int process_notfinished_fds (void);

    int process_ready_fds (void);

    int wait_for_ready_fd (const TimeValue *timeout);
private:
    //
    bool initialized_;
    // The file descriptor associated with the open `/dev/epoll' device.
    int  epoll_fd_;

    // The maximum number of file descriptors over which demultiplexing will occur.
    int size_;

    volatile int deactivated_;

    // Table of event structures to be filled by epoll_wait:
    struct epoll_event *events_;

    // Pointer to the next epoll_event array element that contains the next
    // event to be dispatched.
    struct epoll_event *start_pevents_;

    // The last element in the event array plus one.
    /**
     * The loop that dispatches IO events stops when this->start_pevents_ ==
     * this->end_pevents_.
     */
    struct epoll_event *end_pevents_;

    // Callback object that unblocks the <EpollProactor> if it's sleeping.
    EpollProactorNotify *notify_handler_;

    // Store handle, event-mask, status. !! Just store readable fds's info.
    EpollAioFdInfo *aio_fd_info_;
    typedef RecursiveThreadMutex AIO_FD_INFO_MUTEX_T;
    RecursiveThreadMutex aio_fd_info_mutex_;

    // For not finished list.
    EpollAioFdInfo *not_finished_list_head_;
    EpollAioFdInfo *not_finished_list_tail_;

    // The repository that contains all registered event handlers.
    //EpollAsynchResult *result_list_;

    // Current thread calls <handle_events>
    thread_t owner_;

#if 0
    // The list store the active event triggered by epoll but I/O not completely,
    // as we use Edeg Triggered, if application requests the length of data is 
    // less than the length BUFFER stores, and we recall epoll_wait, it wouldn't
    // Triggered active event, so we must store them in application layout. 
    typedef std::set<NDK_HANDLE> not_finished_fd_list_t;
    typedef std::set<NDK_HANDLE>::iterator not_finished_fd_list_iter;
    not_finished_fd_list_t not_readfinished_list_;
    not_finished_fd_list_t not_writefinished_list_;

    // The list store the results of events.
    //typedef std::deque<EpollAsynchOpt *> result_queue_t;
    //typedef std::deque<EpollAsynchOpt *>::iterator result_queue_iter;
    //result_queue_t result_queue_;
#endif
    EpollProactorToken token_;

    // Write operation task. For emulating aio we need additional threads to 
    // process write operation.
    EpollProactorWriteTask *write_task_;
};

#include "EpollProactor.inl"
#include "Post.h"
#endif

